package cn.carl.frame.concurrent;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

import cn.carl.frame.concurrent.impl.DualImpl;

public class ThreadHelper {
	private static final Logger LOG = Logger.getGlobal();
	public static final int AVAILABLEPROCESSORS = Runtime.getRuntime().availableProcessors();
	private static final ThreadHelper THREADHELPER = new ThreadHelper();
	// 线程池
	public static ExecutorService threadPool;
	// 信号量
	public static ThreadLocal<Semaphore> semaphoreThreadLocal;
	// 锁
	public static ThreadLocal<Lock> lockThreadLocal;

	private ThreadHelper() {
	}

	static {
		threadPool = Executors.newFixedThreadPool(AVAILABLEPROCESSORS * 100);
		lockThreadLocal = new ThreadLocal<>();
		lockThreadLocal.set(new ReentrantLock());
		semaphoreThreadLocal = new ThreadLocal<>();
		//// 根据cpu核心数获取并发,不需要先来后到
		semaphoreThreadLocal.set(new Semaphore(AVAILABLEPROCESSORS * 2, false));
	}

	class ExeInvocationHandler implements InvocationHandler {
		// 目标对象
		private Object target;
		private Boolean shutdown;

		public ExeInvocationHandler(Object target, Boolean shutdown) {
			super();
			this.target = target;
			this.shutdown = shutdown;
		}

		@Override
		public Object invoke(Object arg0, Method method, Object[] param) throws Throwable {
			Object res = null;
			shutdown = shutdown == null ? false : true;
			if (null != param) {
				int size = param.length;
				Semaphore semaphore = getSemaphore();
				semaphore.acquire(size);
				LOG.info(String.format("%s->Get executive signal", Thread.currentThread().getName()));
				res = method.invoke(target, param);
				LOG.info(String.format("%s->Gets and returns all the permissions that are available immediately",
						Thread.currentThread().getName()));
				semaphore.release(size);
				if (shutdown) {
					// 回收线程池
					threadPool.shutdown();
				}
			}
			return res;
		}

		/**
		 * 获取目标对象的代理对象
		 * 
		 * @return Object 代理对象
		 */
		public Object getProxy() {
			return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
					target.getClass().getInterfaces(), this);
		}
	}

	/**
	 * 获取信号量引用
	 * 
	 * @return
	 */
	private static Semaphore getSemaphore() {
		Semaphore semaphore = null;
		if (null == semaphore) {
			semaphore = semaphoreThreadLocal.get();
		}
		return semaphore;
	}

	/**
	 * 获取代理对象
	 * 
	 * @param shutdown
	 *            Boolean 是否关闭线程池
	 * @return Dual 代理对象
	 */
	private static Dual getProxy(Boolean shutdown) {
		Dual dual = new DualImpl();
		// 实例化InvocationHandler
		ExeInvocationHandler invocationHandler = THREADHELPER.new ExeInvocationHandler(dual, shutdown);
		// 根据目标对象生成代理对象
		return (Dual) invocationHandler.getProxy();
	}

	/**
	 * 执行无返回的任务列表
	 * 
	 * @param tasks
	 *            List<Runnable> 无返回任务集合
	 * @param shutdown
	 *            Boolean 是否关闭线程池
	 */
	public static void execute(List<Runnable> tasks, Boolean shutdown) {
		Dual dual = getProxy(shutdown);
		dual.execute(tasks);
	}

	/**
	 * 执行有返回的任务列表
	 * 
	 * @param tasks
	 *            List<Callable> 有返回任务集合
	 * @param shutdown
	 *            Boolean 是否关闭线程池
	 */
	public static List<Future<?>> submit(List<Callable<?>> tasks, Boolean shutdown) {
		Dual dual = getProxy(shutdown);
		return dual.submit(tasks);
	}

	/**
	 * 执行任务
	 * 
	 * @param runnable
	 *            Runnable 线程任务
	 */
	public static void exe(Runnable runnable) {
		threadPool.execute(runnable);
	}

	/**
	 * 提交任务
	 * 
	 * @param task
	 *            Callable<?> 有返回的任务
	 * @return Future<?> 线程返回的结果
	 */
	public static Future<?> sub(Callable<?> task) {
		Future<?> future = threadPool.submit(task);
		return future;
	}
}